package com.shujia.flink.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo2DAtaStreamToTable {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    /**
     *
     * 创建flink  sql流处理的环境
     */
    val table: StreamTableEnvironment = StreamTableEnvironment.create(env)

    /**
     * 1、构建一个流
     *
     */
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))


    /**
     * 2、将一个流转换成一个表
     */

    //在流上定义表--- 动态表 （append only表）
    //Table 相当于sparK sql中的df
    val wordsTable: Table = table.fromDataStream(
      wordsDS, //DataStream
      Schema //指定表结构
        .newBuilder()
        .column("f0", DataTypes.STRING()) //字段名和类型
        .build()
    )

    /**
     * flink sql有两种api
     * 1、SQL
     * 2、DSL---Table   --  基本不用
     *
     */

    /**
     * 3、在表上进行连续查询
     * 连续查询返回一个新的动态表
     *
     * 返回的表有两个情况
     * 1、追加的表
     * 2、更新的表
     *
     */

    //DSL
    /*val countTable: Table = wordsTable
      .groupBy($"f0") //分组
      .select($"f0", $"f0".count()) //聚合*/


    //创建一个临时的视图
    table.createTemporaryView("word", wordsTable)
    /**
     * sqlQuery:的返回值是一个Table的对象，sqlQuery中只能执行查询语句
     *
     * executeSql中可以执行DDl,DQL,DML
     */
    //SQL
    val countTable: Table =  table.sqlQuery(
      """
        |select
        |f0 ,count(1) as c
        |from
        |word
        |group by f0
        |
        |""".stripMargin)


    /**
     * 4、将结果表转换成流
     *
     */
    val countDS: DataStream[Row] = countTable.toChangelogStream

    countDS.print()

    //当在代码中使用到流的时候就需要启动，如果只sql不需要启动
    env.execute()


  }

}
